package cn.tedu.stream.window

import org.apache.flink.api.java.tuple.Tuple
import org.apache.flink.streaming.api.scala.{DataStream, KeyedStream, StreamExecutionEnvironment, WindowedStream}
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow

/**
 * @author Amos
 * @date 2022/5/23
 */

object StreamTimeWindowDemo {
  def main(args: Array[String]): Unit = {
    // 环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    // 构建socket数据源 hello 4  hello 5
    val source: DataStream[String] = env.socketTextStream("hadoop01", 9999)
    // 数据处理
    import org.apache.flink.api.scala._
    val keyedStream: KeyedStream[(String, Int), Tuple] =
      source.flatMap(_.split(" ")).map((_, 1)).keyBy(0)

    // 指定TimeWindow的滚动窗口
//    val windowStream: WindowedStream[(String, Int), Tuple, TimeWindow] =
//      keyedStream.timeWindow(Time.seconds(5))

    // 指定TimeWindow的滑动窗口,每过3秒，统计过去5秒内的数据
    val windowStream: WindowedStream[(String, Int), Tuple, TimeWindow] = keyedStream.timeWindow(Time.seconds(5), Time.seconds(3))

    // 数据聚合
    val result = windowStream.sum(1)
    result.print()
    env.execute()

  }

}
